package com.huaxin.dss2.data.NewSmallStation.fbox.signalr;


import com.github.signalr4j.client.ConnectionState;
import com.github.signalr4j.client.transport.ServerSentEventsTransport;
import com.huaxin.dss2.data.NewSmallStation.fbox.Logger;
import com.huaxin.dss2.data.NewSmallStation.fbox.LoggerFactory;
import com.huaxin.dss2.data.NewSmallStation.fbox.LoginFailedException;
import com.huaxin.dss2.data.NewSmallStation.fbox.TokenManager;
import com.huaxin.dss2.data.NewSmallStation.fbox.signalr.internal.JavaHttpConnection;
import com.huaxin.dss2.util.Constant;
import com.huaxin.dss2.util.ScheduledValue;
import com.huaxin.dss2.util.SignalRUtils;
import java.io.IOException;
import java.net.Proxy;

import com.github.signalr4j.client.hubs.HubConnection;
import com.github.signalr4j.client.hubs.HubProxy;


import java.util.UUID;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public abstract class SignalRConnectionBase {
    private final Semaphore connectEvent;
    private HubConnection hubConnection;
    private HubProxy hubProxy;
    private final Logger logger;
    private final LoggerFactory loggerFactory;
    private final SignalRLoggerWrapper javaconnLogger;
    private final SignalRLoggerWrapper sseLogger;
    private String hubUrl;
    private String signalrClientId;
    private Proxy proxy;
    private TokenManager tokenManager;
    private String accessToken;
    private boolean shouldGetNewToken;
    private int retryCount;
    private long lastConnectedTime = 0;
    private boolean reconnecting;
    private boolean monitorStarted;

    public SignalRConnectionBase(String hubUrl, String signalrClientId, TokenManager tokenManager, Proxy proxy, LoggerFactory loggerFactory) {
        this.hubUrl = hubUrl;
        this.signalrClientId = signalrClientId;
        this.proxy = proxy;
        this.loggerFactory = loggerFactory;
        this.logger = loggerFactory.createLogger(this.getClass());
        this.sseLogger = new SignalRLoggerWrapper(this.loggerFactory.createLogger("ServerSentEventsTransport"));
        this.javaconnLogger = new SignalRLoggerWrapper(this.loggerFactory.createLogger("SignalRTransportConnection"));
        this.tokenManager = tokenManager;
        this.connectEvent = new Semaphore(1);

        Constant.GLOBAL_THREAD.execute(() -> {
            try {
                signalRConnectWorker();
            } catch (InterruptedException e) {
                logger.logError("SignalR connect worker exited with " + e.toString());
                log.error("SignalR连接状态异常，即将进行重新挂起连接",e);
                SignalRUtils.updateServer();
                this.start();
            }
        });

    }

    /**
     * 绑定事件
     */
    private void hookEvents() {
        // 错误
        this.hubConnection.error(this::onConnectionError);
        // 正在连接
        this.hubConnection.reconnecting(this::onReconnecting);
        // 重连
        this.hubConnection.reconnected(this::onReconnected);
        // 关闭
        this.hubConnection.closed(this::onConnectionClosed);
        //创建完成
        this.hubConnection.connected(this::connected);
    }

    /**
     * 解除绑定时间
     */
    private void unHookEvents() {
        // 错误
        this.hubConnection.error(null);
        // 正在连接
        this.hubConnection.reconnecting(null);
        // 重连
        this.hubConnection.reconnected(null);
        // 关闭
        this.hubConnection.closed(null);
        //创建完成
        this.hubConnection.connected(null);
    }

    private void onReconnecting() {
        this.reconnecting = true;
        this.release();
        this.logger.logInformation("Reconnecting. hubconn=" + this.hubConnection.getUrl());
    }

    private void onReconnected() {
        this.reconnecting = false;
        this.logger.logInformation("Reconnected. hubconn=" + this.hubConnection.getUrl());
    }

    private void onConnectionClosed() {
        this.logger.logInformation("Closed. hubconn=" + this.hubConnection.getUrl());
        this.release();
    }

    private void onConnectionError(Throwable e) {
        this.logger.logInformation(String.format("Connection error. exception=%s. hubConnection=%s", e, this.hubConnection.getUrl()));
        if (e != null) {
            Throwable cause = e.getCause();
            if (cause != null && cause.getMessage() != null && cause.getMessage().contains("401")) {
                this.shouldGetNewToken = true;
            }
        }

        if (reconnecting) {
            this.release();
        }
    }

    /**
     * HubProxy代理创建时进行的操作
     * @param hubProxy HubProxy
     */
    protected abstract void onHubProxyCreated(HubProxy hubProxy);

    /**
     * HubProxy代理销毁时进行的操作
     * @param hubProxy HubProxy
     */
    protected abstract void onHubProxyDestroyed(HubProxy hubProxy);

    private void connected() {
        this.reconnecting = false;
        this.logger.logInformation("Connected. hubconn=" + this.hubConnection.getUrl());
    }

    /**
     * 更新token
     * @throws LoginFailedException 登录失败异常
     * @throws InterruptedException 终端异常
     */
    private void updateToken() throws LoginFailedException, InterruptedException {
        int retryCount2 = 0;
        for (; ; ) {
            try {
                this.accessToken = this.tokenManager.getOrUpdateToken(this.accessToken);
                break;
            } catch (LoginFailedException e) {
                throw e;
            } catch (IOException e) {
                logger.logError("Get token failed. " + e.toString());
                int waitTime2 = retryCount2++ * 1000;
                if (waitTime2 > 300000) {
                    waitTime2 = 5000;
                }
                Thread.sleep(waitTime2);
            }
        }
    }

    /**
     * 重连SignalR的方法
     * @update 2018/6/21 实现断网重连
     * @author luqing
     * @throws InterruptedException 抛出异常
     */
    private void signalRConnectWorker() throws InterruptedException {
        for (; ; ) {
            do {
                this.connectEvent.acquire();
            }
            while (this.hubConnection != null && this.hubConnection.getState() == ConnectionState.Connected && !this.reconnecting);

            try {
                updateToken();
                if (this.hubConnection != null) {
                    this.unHookEvents();
                }
                if (this.hubProxy != null) {
                    onHubProxyDestroyed(this.hubProxy);
                }
                if (this.hubConnection != null) {
                    this.hubConnection.disconnect();
                }
                this.hubConnection = new HubConnection(hubUrl, "cid=" + signalrClientId, true, new SignalRLoggerWrapper(loggerFactory.createLogger("SignalR")));
                //厂家提供fbox重连设置
                this.hubConnection.setReconnectOnError(false);
                this.hookEvents();
                this.hubProxy = this.hubConnection.createHubProxy("clienthub");
                this.onHubProxyCreated(this.hubProxy);
                this.hubConnection.setCredentials(request -> request.addHeader("Authorization", "Bearer " + this.accessToken));
                this.connectEvent.drainPermits();
                this.hubConnection.start(new ServerSentEventsTransport(this.sseLogger, new JavaHttpConnection(this.proxy, this.javaconnLogger))).get(600, TimeUnit.SECONDS);
                this.retryCount = 0;
                this.shouldGetNewToken = false;
                this.lastConnectedTime = System.currentTimeMillis();
            } catch (Exception e) {
                assert hubConnection != null;
                assert hubConnection.getState() != ConnectionState.Connected;
                if (e instanceof LoginFailedException) {
                    this.logger.logError("Login error.");
                } else {
                    this.logger.logError("Hub connection start error. " + e.toString());
                    e.printStackTrace();
                }
                Throwable cause = e.getCause();
                if (cause != null) {
                    String msg = cause.getMessage();
                    if (msg != null && msg.contains("401")) {
                        this.shouldGetNewToken = true;
                    }
                }

                int waitTime = this.retryCount++ * 1000;
                if (waitTime > 60000) {
                    waitTime = 5000;
                }

                if (waitTime > 0) {
                    Thread.sleep(waitTime);
                }
                this.release();
                this.reconnecting = true;
            }
        }
    }

    public void start() {
        this.connectEvent.release();
    }

    /**
     * 释放操作
     */
    private void release(){
        this.signalrClientId = UUID.randomUUID().toString();
        //清空开点队列，等待重连推送开点信息
        ScheduledValue.F_BOX_STATE_QUEUE.clear();
        log.info("更新cid进行重连->{}",this.signalrClientId);
        SignalRUtils.createServer(signalrClientId);

        //sdk 重连
        this.connectEvent.release();
    }
}
